// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#pragma once

#include <pollux/dwio/common/range.h>
#include <pollux/dwio/dwrf/writer/statistics_builder.h>
#include <pollux/vector/decoded_vector.h>
#include <pollux/vector/flat_vector.h>

namespace kumo::pollux::dwrf {

class StatisticsBuilderUtils {
 public:
  static void addValues(
      StatisticsBuilder& builder,
      const VectorPtr& vector,
      const common::Ranges& ranges);

  static void addValues(
      BooleanStatisticsBuilder& builder,
      const VectorPtr& vector,
      const common::Ranges& ranges);

  static void addValues(
      BooleanStatisticsBuilder& builder,
      const DecodedVector& vector,
      const common::Ranges& ranges);

  template <typename INT>
  static void addValues(
      IntegerStatisticsBuilder& builder,
      const VectorPtr& vector,
      const common::Ranges& ranges);

  template <typename INT>
  static void addValues(
      IntegerStatisticsBuilder& builder,
      const DecodedVector& vector,
      const common::Ranges& ranges);

  template <typename FLOAT>
  static void addValues(
      DoubleStatisticsBuilder& builder,
      const VectorPtr& vector,
      const common::Ranges& ranges);

  static void addValues(
      StringStatisticsBuilder& builder,
      const VectorPtr& vector,
      const common::Ranges& ranges);

  static void addValues(
      BinaryStatisticsBuilder& builder,
      const VectorPtr& vector,
      const common::Ranges& ranges);
};

template <typename INT>
void StatisticsBuilderUtils::addValues(
    IntegerStatisticsBuilder& builder,
    const VectorPtr& vector,
    const common::Ranges& ranges) {
  auto nulls = vector->raw_nulls();
  auto vals = vector->as_flat_vector<INT>()->rawValues();
  if (vector->may_have_nulls()) {
    for (auto& pos : ranges) {
      if (bits::isBitNull(nulls, pos)) {
        builder.setHasNull();
      } else {
        builder.addValues(vals[pos]);
      }
    }
  } else {
    for (auto& pos : ranges) {
      builder.addValues(vals[pos]);
    }
  }
}

template <typename INT>
void StatisticsBuilderUtils::addValues(
    IntegerStatisticsBuilder& builder,
    const DecodedVector& vector,
    const common::Ranges& ranges) {
  if (vector.may_have_nulls()) {
    for (auto& pos : ranges) {
      if (vector.is_null_at(pos)) {
        builder.setHasNull();
      } else {
        builder.addValues(vector.value_at<INT>(pos));
      }
    }
  } else {
    for (auto& pos : ranges) {
      builder.addValues(vector.value_at<INT>(pos));
    }
  }
}

template <typename FLOAT>
void StatisticsBuilderUtils::addValues(
    DoubleStatisticsBuilder& builder,
    const VectorPtr& vector,
    const common::Ranges& ranges) {
  auto nulls = vector->raw_nulls();
  auto vals = vector->as_flat_vector<FLOAT>()->rawValues();
  if (vector->may_have_nulls()) {
    for (auto& pos : ranges) {
      if (bits::isBitNull(nulls, pos)) {
        builder.setHasNull();
      } else {
        builder.addValues(vals[pos]);
      }
    }
  } else {
    for (auto& pos : ranges) {
      builder.addValues(vals[pos]);
    }
  }
}

} // namespace kumo::pollux::dwrf
